import { cpus } from "node:os";
import { MessageChannel, MessagePort, Worker } from "node:worker_threads";
import { Subject } from "rxjs";
import { dirname, extname } from "node:path";

interface WorkerJob {
  job: "scan";
  value: {
    path: string;
  };
}

export type WorkerMessage =
  | {
      type: "scanResult";
      value: {
        results: Array<{
          path: string;
          isTarget: boolean;
        }>;
      };
    }
  | {
      type: "scan";
      value: {
        path: string;
      };
    }
  | {
      type: "startup";
      value: {
        channel: MessagePort;
        id: number;
      };
    };

export class ScanService {
  // 轮询的当前 worker 的 index。
  private index = 0;
  //   wokers 数组
  private workers: Worker[] = [];
  //   通信用的 port 数组
  private tunnels: MessagePort[] = [];

  startScan(stream$: Subject<string>, path: string) {
    this.initWorkers();
    this.listenEvents(stream$);
    this.addJob({ job: "scan", value: { path } });
  }

  private initWorkers(): void {
    const size = this.getPoolSize();
    // 循环创建 cpu 核心数个 worker，放到 workers 数组里，并创建 channel 放到 tunnels 数组里。
    for (let i = 0; i < size; i++) {
      const { port1, port2 } = new MessageChannel();
      //   这里的 worker 路径是根据当前文件的目录，加上后缀名拼出来的
      // 创建 workers，监听所有 port 的 message 消息，然后给 worker 派发第一个 job
      const worker = new Worker(this.getWorkerPath());
      worker.postMessage(
        {
          type: "startup",
          value: {
            channel: port2,
            id: i,
          },
        },
        [port2]
      );
      this.workers.push(worker);
      this.tunnels.push(port1);
    }
  }

  private getPoolSize() {
    return cpus().length;
  }

  private getWorkerPath() {
    // 在 es module 里，不能用 __dirname，可以换成 import.meta.url
    const actualFilePath = import.meta.url;
    const dirPath = dirname(actualFilePath);
    const extension = extname(actualFilePath);
    const workerName = "scan.worker";
    return new URL(`${dirPath}/${workerName}${extension}`);
  }

  private listenEvents(stream$: Subject<string>) {
    this.tunnels.forEach((tunnel) => {
      tunnel.on("message", (data: WorkerMessage) => {
        this.newWorkerMessage(data, stream$);
      });
    });
  }

  //   返回消息的时候，返回的是 scanResult，带着扫描结果
  private newWorkerMessage(message: WorkerMessage, stream$: Subject<string>) {
    const { type, value } = message;
    if (type === "scanResult") {
      const results: Array<{ path: string; isTarget: boolean }> = value.results;
      results.forEach((result) => {
        const { path, isTarget } = result;
        // 如果 isTarget 是 true，那就找到了目标，用 stream$ 把结果返回给订阅者。
        // 这样，我们只要调用了 startScan 创建 workers 并放第一个 path 进去，线程池就会自己扫描起来，返回的路径只要不是目标路径就会继续用 workers 去扫描。
        if (isTarget) {
          stream$.next(path);
        } else {
          // 否则，就继续用这个 path 去扫描，添加一个 job。
          this.addJob({
            job: "scan",
            value: { path },
          });
        }
      });
    }
  }

  //   派发 job 就是用当前 index 的 worker 来跑任务
  private addJob(job: WorkerJob) {
    if (job.job === "scan") {
      const tunnel = this.tunnels[this.index];
      const message: WorkerMessage = { type: "scan", value: job.value };
      tunnel.postMessage(message);
      this.index = this.index >= this.workers.length - 1 ? 0 : this.index + 1;
    }
  }
}
